package cn.doit.commons

import cn.doit.bean.LogBean
import cn.doit.utils.{Constant, DimensionCombin, JedisUtils}
import org.apache.spark.rdd.RDD
import org.slf4j.LoggerFactory

/**
  * Created by Administrator on 2018/11/16.
  */
object PVAnalys {


  val logger = LoggerFactory.getLogger(this.getClass.getSimpleName)

  /**
    * 实时访问  pv
    *
    * @param filterRdd
    */
  def realFlowPV(filterRdd: RDD[LogBean]) = {

    filterRdd.foreachPartition(fp => {
      lazy val jedis = JedisUtils.getJedis()

      try {
        fp.foreach(fe => {
          val day = fe.getDay
          val hour = fe.getHour
          val minute = fe.getMinute
          val siteId = fe.getSiteId

          val pvRedisKey = "FLOWREAL@PV@" + siteId + "@" + day + "@" + hour + "@" + minute
          jedis.incrBy(pvRedisKey, 1L)

          if (!"1".equals(fe.siteId)) {
            val pvRedisKey_tmp = "FLOWREAL@PV@1" + "@" + day + "@" + hour + "@" + minute
            jedis.incrBy(pvRedisKey_tmp, 1L)
          }
          // 设置key有效期
          //jedis.expire(pvRedisKey, 24 * 60 * 60)
        })
      } catch {
        case e: Exception => {
          logger.error("PVAnalys.realFlowPV foreach error !!!!!!!!!!!!!!!! ")
          e.printStackTrace()
        }
      } finally {
        jedis.close()
      }
    })
  }

  /**
    * 趋势分析PV
    *
    * @param beanRdd
    * @return
    */
  def realTrendPV(beanRdd: RDD[LogBean]) = {

    val filterRdd = beanRdd.flatMap(fe => {
      val day = fe.getDay
      val hour = fe.getHour
      val siteID = fe.getSiteId
      //多维度,设备,省份,访客,来源
      val deviceType = fe.getDeviceType
      val province = fe.getProvince
      val newUser = fe.getNewUser
      val source = fe.getSource
      //与all组合
      val arrDeviceType = Array("all", deviceType)
      val arrProvince = Array("all", province)
      val arrNewUser = Array("all", newUser)
      val arrSource = Array("all", source)
      //拼接维度
      val fieldsList = DimensionCombin.combinKey(arrSource, arrDeviceType, arrProvince, arrNewUser)

      import scala.collection.JavaConversions._
      fieldsList.map(flist => {
        ((flist, day, hour, siteID), 1)
      })
    }).reduceByKey(_ + _)


    filterRdd.foreachPartition(fp => {
      lazy val jedis = JedisUtils.getJedis()
      try {
        fp.foreach {
          case ((flist, day, hour, siteID), num) => {
            //小时
            val tmpHourKey = "FLOWTREND@PV@" + siteID + Constant.symbol + day + Constant.symbol + hour + Constant.symbol + flist
            jedis.incrBy(tmpHourKey, num)
            //天
            val tmpDayKey = "FLOWTREND@PV@" + siteID + Constant.symbol + day + Constant.symbol + "day" + Constant.symbol + flist
            jedis.incrBy(tmpDayKey, num)

            //如果当前站点id为子站点，该子站点的父站点也要累加pv
            if (!"1".equals(siteID)) {
              //小时
              val tmpHourKey = "FLOWTREND@PV@1" + Constant.symbol + day + Constant.symbol + hour + Constant.symbol + flist
              jedis.incrBy(tmpHourKey, num)
              //天
              val tmpDayKey = "FLOWTREND@PV@1" + Constant.symbol + day + Constant.symbol + "day" + Constant.symbol + flist
              jedis.incrBy(tmpDayKey, num)
            }
          }
        }

      } catch {
        case e: Exception => {
          logger.error("PVAnalys.realTrendPV foreach error !!!!!!!!!!!!!!!! ")
          e.printStackTrace()
        }
      } finally {
        logger.info("关闭 jedis 。。。。。。" + jedis)
        jedis.close()
      }
    })
  }

}
